/**
 * 
 */
package com.rrd.dw.mr.key;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

import org.apache.hadoop.io.MapWritable;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.WritableComparator;
import org.apache.hadoop.mapreduce.Partitioner;


/** 
 * @author baoxin.liu
 * @version 创建时间：2019年02月13日 下午7:26:50 
 * 按key正序排序 
 */
public class PairKeyASC implements WritableComparable<PairKeyASC>{
	private String first;
	private String second;
	
	/**
	 * @param second the second to set
	 */
	public void setKey(String left,String right) {
		this.first = left;
		this.second = right;
	}
	
	/**
	 * @return the first
	 */
	public String getFirst() {
		return first;
	}

	
	/**
	 * @return the second
	 */
	public String getSecond() {
		return second;
	}

	

	@Override
	public void write(DataOutput out) throws IOException {
		// TODO Auto-generated method stub
		 out.writeUTF(first);  
//         out.writeLong(second);  
         out.writeUTF(second);
	}

	@Override
	public void readFields(DataInput in) throws IOException {
		first = in.readUTF();
        second = in.readUTF();
	}

	@Override
	public int compareTo(PairKeyASC o) {
        if (!first.equals(o.first)) {  
            return first.compareTo(o.first)>0? 1 : -1; 
        } else if (!second.equals(o.second)) {  
            return second.compareTo(o.second)>=0 ? 1 : -1;  
        } else {  
            return 0;  
        }  
	}
	
	/* (non-Javadoc)
	 * @see java.lang.Object#hashCode()
	 */
	@Override
	public int hashCode() {
		// TODO Auto-generated method stub
		return (int) (first.hashCode()*127);
	}
	/* (non-Javadoc)
	 * @see java.lang.Object#equals(java.lang.Object)
	 */
	@Override
	public boolean equals(Object right) {
		// TODO Auto-generated method stub
		if (right == null)  
            return false;  
        if (this == right)  
            return true;  
        if (right instanceof PairKeyASC) {  
        	PairKeyASC r = (PairKeyASC) right;  
            return r.first.equals(this.first) && r.second.equals(this.second);  
        } else {  
            return false;  
        }  
	}
	
	public static class FirstPartitioner extends Partitioner<PairKeyASC,MapWritable>{

		/* (non-Javadoc)
		 * @see org.apache.hadoop.mapreduce.Partitioner#getPartition(java.lang.Object, java.lang.Object, int)
		 */
		@Override
		public int getPartition(PairKeyASC key, MapWritable value, int numPartitions) {
			// TODO Auto-generated method stub
			return Math.abs(key.hashCode()) % numPartitions;  
		}

	}
	public static class MyGroupingComparator extends WritableComparator{

		public MyGroupingComparator(){
			super(PairKeyASC.class,true);
		}
		@Override
		public int compare(Object a, Object b) {
			// TODO Auto-generated method stub
			 PairKeyASC ip1 = (PairKeyASC) a;  
	    	   PairKeyASC ip2 = (PairKeyASC) b;  
	         String l = ip1.getFirst();  
	         String r = ip2.getFirst();  
	         return l.compareTo(r);  
		}
 
	}
}

